package com.example.demo.mqclient;

import com.example.demo.common.*;
import com.example.demo.mqserver.core.BasicProperties;
import com.example.demo.mqserver.core.Exchange;
import com.example.demo.mqserver.core.ExchangeType;
import jdk.nashorn.internal.runtime.regexp.joni.ast.StringNode;
import lombok.Data;

import javax.websocket.RemoteEndpoint;
import java.io.IOException;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;

/**
 * Created with IntelliJ IDEA.
 * Description:
 * User: 张
 * Date: 2024-08-24
 * Time: 3:03
 */
@Data
public class Channel {
    private String channelId;
    // 当前的channel是属于哪个连接的
    private Connection connection;
    // 用来存储客户端后续接收到的服务器的响应 --- 我们需要读取socket中收到的所有的响应数据,这张表就是存储响应数据的
    private ConcurrentHashMap<String, BasicReturn> basicReturnMap = new ConcurrentHashMap<>();
    //此处约定,一个channel中只能有一个回调
    private Consumer consumer = null;//客户端后续接收到消息时,所执行的回调

    public Channel(String channelId, Connection connection) {
        this.channelId = channelId;
        this.connection = connection;
    }

    //和BrokerServer进行交互,告知服务器创建Channel消息
    public boolean createChannel() throws IOException {
        //创建Channel中的payload就是一个BasicArguments对象即可
        BasicArguments basicArguments = new BasicArguments();
        basicArguments.setChannelId(channelId);
        basicArguments.setRid(generateRid());
        byte[] payload = BinaryTool.toBytes(basicArguments);
        Request request = new Request();
        request.setType(0x1);
        request.setLength(payload.length);
        request.setPayload(payload);
        //发送请求
        connection.writeRequest(request);
        //等待服务器响应
        BasicReturn basicReturn = waitResult(basicArguments.getRid());//阻塞等待响应
        System.out.println("[Channel] 成功等待到响应!! rid = " + basicReturn.getRid());
        return basicReturn.isOk();
    }

    //使用这个方法在进行响应的阻塞等待
    private BasicReturn waitResult(String rid) {
        //反复查询 basicReturnMap,查看是否有对应的rid响应
        BasicReturn basicReturn = null;
        while ((basicReturn = basicReturnMap.get(rid)) == null) {
            //执行结果为空,说明其对应响应还没回
            //此时就进行阻塞等待
            synchronized (this) {
                try {
                    wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
        // 读取成功之后要把该条响应数据从hash表中删除
        basicReturnMap.remove(rid);
        return basicReturn;
    }

    //添加响应
    public void putReturn(BasicReturn basicReturn) {
        System.out.println("[Channel] 开始添加响应! basicReturn = " + basicReturn.toString());
        basicReturnMap.put(basicReturn.getRid(),basicReturn);
        System.out.println("[Channel] 添加响应成功!!! rid = " + basicReturn.getRid());
        //唤醒等待的线程
        synchronized (this) {
            //当前不知道有多少个线程在等待上述的响应,直接把所有等待的线程都唤醒出来认一下"人"
            //差不到的就接着等
            notify();
        }
    }

    private String generateRid() {
        return "C-" + UUID.randomUUID().toString();
    }

    //关闭channel type = 0x2
    public boolean close() throws IOException {
        BasicArguments basicArguments = new BasicArguments();
        basicArguments.setRid(generateRid());
        basicArguments.setChannelId(channelId);
        byte[] payload = BinaryTool.toBytes(basicArguments);

        Request request = new Request();
        request.setType(0x2);
        request.setLength(payload.length);
        request.setPayload(payload);

        //发送请求
        connection.writeRequest(request);
        //等响应
        BasicReturn basicReturn = waitResult(basicArguments.getRid());
        return basicReturn.isOk();
    }

    //创建交换机
    public boolean exchangeDeclare(String exchangeName, ExchangeType exchangeType, boolean durable, boolean autoDelete,
                                   Map<String, Object> arguments) throws IOException {
        ExchangeDeclareArguments exchangeDeclareArguments = new ExchangeDeclareArguments();
        exchangeDeclareArguments.setRid(generateRid());
        exchangeDeclareArguments.setChannelId(channelId);
        exchangeDeclareArguments.setExchangeName(exchangeName);
        exchangeDeclareArguments.setExchangeType(exchangeType);
        exchangeDeclareArguments.setArguments(arguments);
        exchangeDeclareArguments.setAutoDelete(autoDelete);
        byte[] payload = BinaryTool.toBytes(exchangeDeclareArguments);

        Request request = new Request();
        request.setType(0x3);
        request.setLength(payload.length);
        request.setPayload(payload);
        //发送请求
        connection.writeRequest(request);
        //等
        BasicReturn basicReturn = waitResult(exchangeDeclareArguments.getRid());
        return basicReturn.isOk();
    }

    //删除交换机
    public boolean exchangeDelete(String exchangeName) throws IOException {
        ExchangeDeleteArguments arguments = new ExchangeDeleteArguments();
        arguments.setRid(generateRid());
        arguments.setChannelId(channelId);
        arguments.setExchangeName(exchangeName);
        byte[] payload = BinaryTool.toBytes(arguments);

        Request request = new Request();
        request.setType(0x4);
        request.setLength(payload.length);
        request.setPayload(payload);
        //发送请求
        connection.writeRequest(request);
        //等
        BasicReturn basicReturn = waitResult(arguments.getRid());
        return basicReturn.isOk();
    }

    //创建队列
    public boolean queueDeclare(String queueName, boolean durable, boolean exclusive, boolean autoDelete,
                                Map<String, Object> arguments) throws IOException {
        QueueDeclareArguments queueDeclareArguments = new QueueDeclareArguments();
        queueDeclareArguments.setRid(generateRid());
        queueDeclareArguments.setChannelId(channelId);
        queueDeclareArguments.setQueueName(queueName);
        queueDeclareArguments.setDurable(durable);
        queueDeclareArguments.setExclusive(exclusive);
        queueDeclareArguments.setAutoDelete(autoDelete);
        queueDeclareArguments.setArguments(arguments);
        byte[] payload = BinaryTool.toBytes(queueDeclareArguments);

        Request request = new Request();
        request.setType(0x5);
        request.setLength(payload.length);
        request.setPayload(payload);

        connection.writeRequest(request);
        BasicReturn basicReturn = waitResult(queueDeclareArguments.getRid());
        return basicReturn.isOk();
    }

    //销毁队列
    public boolean queueDelete(String queueName) throws IOException {
        QueueDeleteArguments arguments = new QueueDeleteArguments();
        arguments.setRid(generateRid());
        arguments.setChannelId(channelId);
        arguments.setQueueName(queueName);
        byte[] payload = BinaryTool.toBytes(arguments);

        Request request = new Request();
        request.setType(0x6);
        request.setLength(payload.length);
        request.setPayload(payload);

        connection.writeRequest(request);
        BasicReturn basicReturn = waitResult(arguments.getRid());
        return basicReturn.isOk();
    }

    //创建绑定
    public boolean queueBind(String queueName, String exchangeName, String bindingKey) throws IOException {
        QueueBindArguments arguments = new QueueBindArguments();
        arguments.setRid(generateRid());
        arguments.setChannelId(channelId);
        arguments.setExchangeName(exchangeName);
        arguments.setQueueName(queueName);
        arguments.setBindingKey(bindingKey);
        byte[] payload = BinaryTool.toBytes(arguments);

        Request request = new Request();
        request.setType(0x7);
        request.setLength(payload.length);
        request.setPayload(payload);

        connection.writeRequest(request);
        BasicReturn basicReturn = waitResult(arguments.getRid());
        return basicReturn.isOk();
    }

    //解除绑定
    public boolean queueUnBind(String queueName, String exchangeName) throws IOException {
        QueueUnbindArguments arguments = new QueueUnbindArguments();
        arguments.setRid(generateRid());
        arguments.setChannelId(channelId);
        arguments.setExchangeName(exchangeName);
        arguments.setQueueName(queueName);
        byte[] payload = BinaryTool.toBytes(arguments);

        Request request = new Request();
        request.setType(0x8);
        request.setLength(payload.length);
        request.setPayload(payload);

        connection.writeRequest(request);
        BasicReturn basicReturn = waitResult(arguments.getRid());
        return basicReturn.isOk();
    }

    //发送消息 --- 生产者往队列中加入消息(先发交换机)
    public boolean basicPublish(String exchangeName, String routingKey, BasicProperties basicProperties, byte[] body) throws IOException {
        BasicPublishArguments arguments = new BasicPublishArguments();
        arguments.setRid(generateRid());
        arguments.setChannelId(channelId);
        arguments.setExchangeName(exchangeName);
        arguments.setRoutingKey(routingKey);
        arguments.setBasicProperties(basicProperties);
        arguments.setBoyd(body);
        byte[] payload = BinaryTool.toBytes(arguments);

        Request request = new Request();
        request.setType(0x9);
        request.setLength(payload.length);
        request.setPayload(payload);

        connection.writeRequest(request);
        BasicReturn basicReturn = waitResult(arguments.getRid());
        return basicReturn.isOk();
    }

    //订阅消息
    public boolean basicConsume(String queueName, boolean autoAck, Consumer consumer) throws MqException, IOException {
        //先设置回调
        if (this.consumer != null) {
            //一个channel只能有一种回调
            throw new MqException("该Channel 已经设置了消费消息的回调函数了,不能重复设置!!!");
        }
        this.consumer = consumer;
        BasicConsumeArguments arguments = new BasicConsumeArguments();
        arguments.setRid(generateRid());
        arguments.setChannelId(channelId);
        arguments.setConsumerTag(channelId);//此处的consumerTag使用channelId来设置
        arguments.setQueueName(queueName);
        arguments.setAutoAck(autoAck);
        byte[] payload = BinaryTool.toBytes(arguments);

        Request request = new Request();
        request.setType(0xa);
        request.setLength(payload.length);
        request.setPayload(payload);

        connection.writeRequest(request);
        BasicReturn basicReturn = waitResult(arguments.getRid());
        return basicReturn.isOk();
    }

    //确认消息
    public boolean basicAck(String queueName, String messageId) throws IOException {
        BasicAckArguments arguments = new BasicAckArguments();
        arguments.setRid(generateRid());
        arguments.setChannelId(channelId);
        arguments.setQueueName(queueName);
        arguments.setMessageId(messageId);
        byte[] payload = BinaryTool.toBytes(arguments);

        Request request = new Request();
        request.setType(0xb);
        request.setLength(payload.length);
        request.setPayload(payload);

        connection.writeRequest(request);
        BasicReturn basicReturn = waitResult(arguments.getRid());
        return basicReturn.isOk();
    }


}



